﻿/***********************************************************************
** Author       : 陈涛
** Date         : 2012-9-1
** Name         : rec_task.cpp
** Version      : 1.0
** Description  :
   该类的功能，主要是实现录像管理线程。
** Modify Record:
2015年5月29日11时16分33秒：去除之前的每隔30秒刷新磁盘的操作
***********************************************************************/
#include "rec_data_io.h"
#include "cross_com.h"
//录像管理类
/*
	开启一个线程去录像针对一个硬盘
	使用了MAP容器，可以保证录像文件在磁盘上面位置是从小到大 LBA 排序，有效解决磁盘寻道的时候
	写入数据关键思想，避免再次拷贝内存，减少CPU占用率。将FIFO的内存地址给出
	大致流程
	1. 判断录像是否结束，如果结束，需要将当前录像文件的FIFO缓冲当中的数据全部刷新到磁盘。
	2. 刷新I帧索引信息
	3. 写入录像数据

*/

//#define TIME_LOOK

uint32_t CRecordDataIO::ST_WT_WATER = 30;
uint32_t CRecordDataIO::ST_RD_WATER = 45;			//
const int DEF_WAIT_COUNT = 10;

CDiskRecordInfo::CDiskRecordInfo()
{
	thd_start = false;
	wait = false;
	replay_num = false;
	rpBuf = NULL;
	rpBufSize = 0;
	thd_id   = 0;
	full_lba = 0;
	memset(device,0,24);
	memset(storage_name,0,24);
}
int  CRecordDataIO::DataFlushThread(void *pParam)
{
	CDiskRecordInfo * pTask = (CDiskRecordInfo * ) pParam;
	TRACE("CRecordDataIO::DataFlushThread.\n");
	pTask->RecordFlushProc();

	return 0;
}

int  CRecordDataIO::ResizeRpBuf(char * device,int size)
{
	for(int i = 0 ; i < (int)m_vec_task.size() ; i++)
	{
	//	if(device == m_vec_task[i]->dev)
		{
			if (size != (int) m_vec_task[i]->rpBufSize)
			{
				delete m_vec_task[i]->rpBuf;
				m_vec_task[i]->rpBuf = new uchar_t[size];
				m_vec_task[i]->rpBufSize = size;
			}
			return 0;
		}
	}
	return ERR_DATA_BUF_EMPYT;
}

int CDiskRecordInfo::RecordFlushProc()
{
	CRecWriteBase * pRec = NULL;
	 CRecReadBase * pReader = NULL;

	MAP_RECORD_READER::iterator it_reader,it_reader_bak;

	MAP_RECORD_WRITER::iterator it_rec,it_rec_bak;
	MAP_RECORD_WRITER::reverse_iterator  it_rrec;

	uint32_t cache_water = 30;
	const uint32_t max_water = 60;					//默认最高水位，只要到这个水位，一定需要泄流

	int cache_time = 3000;					//FIFO缓冲保留的最长时间 单位ms
	int ret = 0;

#ifdef TIME_LOOK
	cross_time_t tm_start,tm_stop;
#endif

	int32_t wait_count     = DEF_WAIT_COUNT;					//默认10次
	int     sleep_all_ch   = 50;

	TRACE("DevRec thread start. sleep time rec %d.\n",sleep_all_ch);

	//录像循环线程 此思路对于车载DVR不适合，数据延时太久写入会造成意外断电后丢失的录像过多。
	/*
		关键思想
		1.磁盘文件按LBA从小到大顺序，去写入，理论上可以最大程度的减少磁盘大范围跳转所带来延时
		2.录像数据写入均按磁盘扇区的整数倍写入，减少因为不对齐所带来的 读取--修改--写入
		3.线程最后进行一些延时，分出一些时间片来给其上层线程来拷贝数据到FIFO缓冲，因为磁盘的写入数据的速度远大于16路录像通道所产生的数据量。
		4.定时刷新I帧索引
		5.对于写入出错的通道，应该及时删除当前通道录像。
		6.只要上层停止录像，就应该立即将数据刷新到磁盘上面
		7.时间超过5000ms就必须强制写入数据
	*/

	while(true)
	{
		if(((int)mp_rec.size() == 0) && ((int)mp_recRead.size() ==0))
		{
			CrossSleep(1000); //
			thd_start = false;

			continue;
		}

		thd_start = true;

#ifdef TIME_LOOK
		TIME_CNT_CURRENT(tm_start);
#endif
        devRecLock.Lock();
		for(it_rec = mp_rec.begin(); it_rec !=  mp_rec.end(); )
		{
			pRec = it_rec->second;				//获取录像文件类

			ret = pRec->Flush_FIFOData(cache_time,cache_water);
			if(ret < 0)
			{
				//错误值处理
				TRACE_ERR("Flush_FIFOData Error ret %d error %d ",ret,errno);
				it_rec_bak = it_rec;
				it_rec++;
				mp_rec.erase(it_rec_bak);
				continue;
			}
			it_rec++;
		}
		devRecLock.UnLock();

		//Replay download read func
		devReplayLock.Lock();
		if(0 != mp_recRead.size())
		{
			for(it_reader = mp_recRead.begin();it_reader != mp_recRead.end();)
			{
				pReader = it_reader->second;
				if(!pReader->IsOpen() || (it_reader->first != pReader->GetLoopLBA()))
				{
					TRACE("CH%02d index %d  frame index num %d close file.\n",pReader->GetChannel(), pReader->GetFileIndex(),pReader->GetIndexNum());
					it_reader_bak = it_reader;
					it_reader++;
					mp_recRead.erase(it_reader_bak);
					continue;
				}

				if(it_reader->first == pReader->GetLoopLBA())
				{
					ret     = pReader->Read_TO_FIFO(CRecordDataIO::ST_RD_WATER);
					if(ret < 0)
					{
						pReader->Close();
						TRACE("CH%02d Read_TO_FIFO Error ret %d error %d \n",
                                   pReader->GetChannel(),ret,errno);
						it_reader_bak = it_reader;
						it_reader++;
						mp_recRead.erase(it_reader_bak);
						continue;
					}
				}
				it_reader++;
			}

		}
		devReplayLock.UnLock();
		CrossSleep(sleep_all_ch);
		//对于车载录像机，实时性更高，一定要缩短数据写入的时间。

		wait_count = DEF_WAIT_COUNT;
	}

	thd_id = (cross_thread_t)0;
	TRACE("DevRec thread END.\n");
	return 0;
}

int CRecordDataIO::AddRecordTask(char* pStorage,uint64_t lba,CRecWriteBase* pRec)		//增加一个录像任务
{
	//检查是不是新的磁盘设备
	CDiskRecordInfo* dev_rec_mp = NULL;
	CDiskRecordInfo* pdev_rec_mp = NULL;
	MAP_RECORD_WRITER::iterator it_rec;
	VEC_DISK_REC_INFO::iterator it_dev;
	char tmpStorage[80] = {0};

	for(it_dev = m_vec_task.begin() ;it_dev != m_vec_task.end();it_dev++)
	{
		if(strstr(pStorage,((CDiskRecordInfo*)*it_dev)->storage_name))
		{
//			TRACE("Find device  %s \n",pStorage);
			pdev_rec_mp =(CDiskRecordInfo*) (*it_dev);
			goto ADD_RECORD_FILE ;
		}
	}

	dev_rec_mp = new CDiskRecordInfo();
	dev_rec_mp->devRecLock.Lock();

	GetStorageName(tmpStorage,pStorage);
	strcpy(dev_rec_mp->storage_name,tmpStorage);

	dev_rec_mp->thd_start =false;
	dev_rec_mp->mp_rec[lba] = pRec;
	dev_rec_mp->full_lba = 0;
	m_vec_task.push_back(dev_rec_mp);
	//未找到的情况下，需要创建新的线程去写入录像
	TRACE_ERR("Create thread for record.\n");
	pdev_rec_mp	= dev_rec_mp;

	if(pdev_rec_mp->thd_id == PUB_CREATE_THREAD_FAIL)
	{
		pdev_rec_mp->thd_id = Cross_CreateThread((cross_start_routine)DataFlushThread, (void *)pdev_rec_mp, NULL);
	}
	pdev_rec_mp->wait = true;
	pdev_rec_mp->replay_num = 0;
	dev_rec_mp->devRecLock.UnLock();

	return 1;

ADD_RECORD_FILE:
	//开启新录像加入到录像列表当中
	pdev_rec_mp->devRecLock.Lock();
	pdev_rec_mp->thd_start  = true;
	pdev_rec_mp->mp_rec[lba] = pRec;
	pdev_rec_mp->devRecLock.UnLock();

#if 1
	if(PUB_CREATE_THREAD_FAIL == pdev_rec_mp->thd_id)
	{	//假如线程出现退了情况，则需要重新启动刷新线程。
		TRACE("create thread.  pdev_rec_mp thd_id %d.\n", (int)pdev_rec_mp->thd_id);
	//	TRACE_ERR("Re-cread thhread to flush fifo data. dev %s \n",pdev_rec_mp->dev->get_device_name());
		pdev_rec_mp->thd_id = Cross_CreateThread((cross_start_routine)DataFlushThread, (void *)pdev_rec_mp, NULL);
		pdev_rec_mp->wait = true;
		pdev_rec_mp->replay_num = 0;
	}
#endif

#if 1
	TRACE("Add record.Record file indexNo %d, Record num %d   \n",pRec->GetRecIdxNo(),(int)pdev_rec_mp->mp_rec.size());
	if((int)pdev_rec_mp->mp_rec.size() && pRec->get_channel() == 0 )
	{
		for(it_rec = pdev_rec_mp->mp_rec.begin() ;it_rec !=  pdev_rec_mp->mp_rec.end(); it_rec++)
		{
			TRACE("CH %02d LBA %lld \n",it_rec->second->get_channel(),it_rec->second->get_lba_start());
		}
	}
#endif

	return 0;
}

int CRecordDataIO::DelRecordTask(char* pStorage,uint64_t lba,CRecWriteBase* pRec)		//删除录像任务
{
	VEC_DISK_REC_INFO::iterator it_devTask;
	MAP_RECORD_WRITER::iterator it_rec;

	for(it_devTask = m_vec_task.begin(); it_devTask != m_vec_task.end();it_devTask++)
	{
		if(strstr(((CDiskRecordInfo*)*it_devTask)->storage_name,(char*)pStorage))
		{
			(*it_devTask)->devRecLock.Lock();

			MAP_RECORD_WRITER* pMP_rec = &((*it_devTask))->mp_rec;
			for (it_rec = pMP_rec->begin();it_rec != pMP_rec->end(); it_rec++)
			{
				if((lba == it_rec->first) && (it_rec->second == pRec))
				{
					TRACE_ERR("Found CH %d the rec-file.LBA %lld cur-offset %u \n",pRec->get_channel(),pRec->get_lba_start(),pRec->GetCurPos());
					pMP_rec->erase(it_rec);
					break;
				}
			}

			(*it_devTask)->devRecLock.UnLock();
			return 0;
		}
	}

	return -1;
}
int CRecordDataIO::AddReaderTask(char* pStorage,uint64_t lba,CRecReadBase* pRec)
{
	//检查是不是新的磁盘设备
	TRACE("Start AddReadTask storage %s pStorage lba %lld. size %d.\n",pStorage ,lba,m_vec_task.size());
	CDiskRecordInfo* dev_rec_mp  = NULL;
	CDiskRecordInfo* pdev_rec_mp = NULL;
	VEC_DISK_REC_INFO::iterator it_dev;

	for(it_dev = m_vec_task.begin() ;it_dev != m_vec_task.end();it_dev++)
	{
		if(strstr(pStorage,((CDiskRecordInfo*)*it_dev)->storage_name))
		{
			pdev_rec_mp =(CDiskRecordInfo*) (*it_dev);
			if(pdev_rec_mp->rpBuf == NULL)
			{
				pdev_rec_mp->rpBuf = new uchar_t[RP_ALIGNBUF_SIZE];
				pdev_rec_mp->rpBufSize = RP_ALIGNBUF_SIZE;
			}
			TRACE("Storage %s Start  lba %lld addReaderTask found.\n",pStorage ,lba);
			goto ADD_RECORD_FILE ;
		}
	}
	dev_rec_mp = new CDiskRecordInfo();

	strcpy(dev_rec_mp->storage_name,pStorage);
	dev_rec_mp->thd_start =false;
	dev_rec_mp->mp_recRead[lba] = pRec;
	dev_rec_mp->full_lba   = 0;
	dev_rec_mp->rpBuf      = new uchar_t[RP_ALIGNBUF_SIZE];
	dev_rec_mp->rpBufSize  = RP_ALIGNBUF_SIZE;
	m_vec_task.push_back(dev_rec_mp);
	//未找到的情况下，需要创建新的线程去写入录像

	pdev_rec_mp	= dev_rec_mp;
	pdev_rec_mp->thd_id = Cross_CreateThread((cross_start_routine)DataFlushThread, (void *)pdev_rec_mp, NULL);
	pdev_rec_mp->wait = true;
	pdev_rec_mp->replay_num +=1;

	return 1;

ADD_RECORD_FILE:
	//开启新录像加入到录像列表当中
	pdev_rec_mp->devReplayLock.Lock();
	pdev_rec_mp->thd_start       = true;
	pdev_rec_mp->mp_recRead[lba] = pRec;
	pdev_rec_mp->devReplayLock.UnLock();

	if(PUB_CREATE_THREAD_FAIL == pdev_rec_mp->thd_id)
	{	//假如线程出现退了情况，则需要重新启动刷新线程。
		TRACE("Re-cread thhread to flush fifo data. dev %s \n",pdev_rec_mp->device);
		pdev_rec_mp->thd_id = Cross_CreateThread((cross_start_routine)DataFlushThread, (void *)pdev_rec_mp, NULL);
		pdev_rec_mp->wait = true;

	}
	pdev_rec_mp->replay_num +=1;

	return 0;
}

int CRecordDataIO::DelReaderTask(char* pStorage,uint64_t lba,CRecReadBase* pRec)
{
	VEC_DISK_REC_INFO::iterator it_devTask;

	for(it_devTask = m_vec_task.begin(); it_devTask != m_vec_task.end();it_devTask++)
	{
		if(strstr(((*it_devTask))->storage_name , pStorage))
		{
			(*it_devTask)->devReplayLock.Lock();
			MAP_RECORD_READER::iterator it_rec;
			MAP_RECORD_READER* pMP_rec = &((*it_devTask))->mp_recRead;
			for (it_rec = pMP_rec->begin();it_rec != pMP_rec->end(); it_rec++)
			{
				TRACE("DelReaderTask cur LBA %lld map Key %lld .\n",lba,it_rec->first);
				if((lba == it_rec->first) && (it_rec->second == pRec))
				{
					TRACE_ERR("Found CH %d the reader file.LBA %lld .\n",pRec->GetChannel(),pRec->GetLoopLBA());
					PrintTimeMs(pRec->GetStartTime(),pRec->GetEndTime(),"Close record time");
					pMP_rec->erase(it_rec);
					break;
				}
			}

			(*it_devTask)->devReplayLock.UnLock();
			return 0;
		}
	}

	return -1;
}
